package com.changan;

import com.changan.common.model.Tuple2;
import com.changan.config.ColumnEntity;
import com.changan.config.EnvConfig;
import com.changan.config.TableEntity;
import com.changan.model.AppArgs;
import com.changan.parser.CliTool;
import com.changan.services.FlinkService;
import com.changan.utils.MysqlUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class StreamApp {

    private static final String CDC_PREFIX = "cdc_";
    private static final String STARROCKS_PREFIX = "starrocks_";
    private static final String STREAM_DATA_VIEW_SUFFIX = "_data_stream_view";
    private static final String STREAM_SCHEMA_VIEW_SUFFIX = "_schema_stream_view";
    private static final String OUTPUT_TAG_PREFIX = "output_";
    private static final String OUTPUT_TAG_SUFFIX = "_ops";
    private static final String PIPELINE_KEY = "pipeline.name";
    private final static Logger LOG = LoggerFactory.getLogger(StreamApp.class);

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.milliseconds(10 * 1000)));
        StreamTableEnvironment streamTableEnv = getStreamTableEnv(env);
        Configuration configuration = streamTableEnv.getConfig().getConfiguration();

        AppArgs appArgs = CliTool.getJcommander(AppArgs.class, args);

        LOG.info("System accepted the start argument: {}", appArgs);
        List<Tuple2<TableEntity, List<ColumnEntity>>> sources = FlinkService.getFlinkTaskSourceDefinitions(appArgs);
        List<Tuple2<TableEntity, List<ColumnEntity>>> targets = FlinkService.getFlinkTaskTargetDefinitions(appArgs, sources);

        //设置jobName
        setJobName(args, configuration);

        StatementSet statementSet = streamTableEnv.createStatementSet();
        for (int i = 0; i < sources.size(); i++) {
            Tuple2<TableEntity, List<ColumnEntity>> source = sources.get(i);
            Tuple2<TableEntity, List<ColumnEntity>> target = targets.get(i);

            String sourceViewName = CDC_PREFIX + source._1.getQualifier();
            String kafkaSql = FlinkService.getDebeziumSql(appArgs, source._1, source._2, sourceViewName);
            trackThings(kafkaSql);
            streamTableEnv.executeSql(kafkaSql);

            String sinkViewName = STARROCKS_PREFIX + target._1.getQualifier();
            String starrocksSql = FlinkService.getStarrocksSql(appArgs, target._1, target._2, sinkViewName);
            trackThings(starrocksSql);
            streamTableEnv.executeSql(starrocksSql);

            String insertSql = insertSql(sourceViewName, sinkViewName);
            trackThings(insertSql);
            statementSet.addInsertSql(insertSql);
            trackThings("=============================================");
        }
        statementSet.execute();
    }

    public static StreamTableEnvironment getStreamTableEnv(StreamExecutionEnvironment env) {
        EnvironmentSettings streamSetting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        return StreamTableEnvironment.create(env, streamSetting);
    }

    public static void trackThings(String... thing) {
        for (String s : thing) {
            LOG.info("\n STREAM_APP: {}", s);
        }
    }

    public static String insertSql(String sourceTable, String targetTable) {
        return "INSERT INTO " + targetTable + " SELECT * FROM " + sourceTable;
    }

    public static void checkStartArgs(String[] args) {
        if (args.length < 1) {
            throw new IllegalArgumentException("You didn't pass any arguments to start, plz check that!");
        }
    }

    public static void setJobName(String[] args, Configuration configuration) {
        if (args.length >= 2) {
            configuration.setString(PIPELINE_KEY, args[1]);
        }
    }
}
